// 模块间通信系统
use std::collections::HashMap;
use std::sync::{Arc, RwLock};
use tokio::sync::broadcast;
use tokio_stream::wrappers::BroadcastStream;
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use futures::StreamExt;
use reqwest;

use crate::modules::interfaces::*;

// 消息总线
#[derive(Debug, Clone)]
pub struct MessageBus {
    channels: Arc<RwLock<HashMap<String, broadcast::Sender<ModuleMessage>>>>,
}

impl MessageBus {
    pub fn new() -> Self {
        Self {
            channels: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    // 创建或获取消息通道
    pub fn channel(&self, name: &str) -> broadcast::Receiver<ModuleMessage> {
        let mut channels = self.channels.write().unwrap();
        if let Some(sender) = channels.get(name) {
            sender.subscribe()
        } else {
            let (sender, receiver) = broadcast::channel(100);
            channels.insert(name.to_string(), sender);
            receiver
        }
    }

    // 发布消息到指定通道
    pub fn publish(&self, channel: &str, message: ModuleMessage) -> Result<usize, broadcast::error::SendError<ModuleMessage>> {
        let channels = self.channels.read().unwrap();
        if let Some(sender) = channels.get(channel) {
            sender.send(message)
        } else {
            Err(broadcast::error::SendError(message))
        }
    }

    // 广播消息到所有通道
    pub fn broadcast(&self, message: ModuleMessage) -> usize {
        let channels = self.channels.read().unwrap();
        let mut total_sent = 0;
        
        for sender in channels.values() {
            if let Ok(_) = sender.send(message.clone()) {
                total_sent += 1;
            }
        }
        
        total_sent
    }
}

// RPC客户端
#[derive(Debug, Clone)]
pub struct RpcClient {
    base_url: String,
    client: reqwest::Client,
}

impl RpcClient {
    pub fn new(base_url: String) -> Self {
        Self {
            base_url,
            client: reqwest::Client::new(),
        }
    }

    // 调用远程方法
    pub async fn call<T: Serialize + ?Sized, R: for<'de> Deserialize<'de>>(
        &self,
        method: &str,
        params: &T,
    ) -> Result<ApiResponse<R>, reqwest::Error> {
        let url = format!("{}/{}", self.base_url, method);
        let response = self.client
            .post(&url)
            .json(params)
            .send()
            .await?
            .json::<ApiResponse<R>>()
            .await?;

        Ok(response)
    }

    // 调用用户管理模块
    pub async fn get_user(&self, user_id: i32) -> Result<ApiResponse<UserInterface>, reqwest::Error> {
        self.call("user/get", &serde_json::json!({ "user_id": user_id })).await
    }

    pub async fn create_user(&self, user_data: UserInterface) -> Result<ApiResponse<UserInterface>, reqwest::Error> {
        self.call("user/create", &user_data).await
    }

    // 调用网关服务模块
    pub async fn get_route_rules(&self) -> Result<ApiResponse<Vec<RouteRuleInterface>>, reqwest::Error> {
        self.call("gateway/routes", &serde_json::json!({})).await
    }

    pub async fn create_route_rule(&self, rule_data: RouteRuleInterface) -> Result<ApiResponse<RouteRuleInterface>, reqwest::Error> {
        self.call("gateway/route/create", &rule_data).await
    }

    // 调用菜单管理模块
    pub async fn get_menus(&self) -> Result<ApiResponse<Vec<MenuInterface>>, reqwest::Error> {
        self.call("menu/list", &serde_json::json!({})).await
    }

    pub async fn create_menu(&self, menu_data: MenuInterface) -> Result<ApiResponse<MenuInterface>, reqwest::Error> {
        self.call("menu/create", &menu_data).await
    }
}

// RPC服务器
pub struct RpcServer {
    message_bus: Arc<MessageBus>,
}

impl RpcServer {
    pub fn new(message_bus: Arc<MessageBus>) -> Self {
        Self { message_bus }
    }

    // 处理RPC请求
    pub async fn handle_request(&self, method: &str, params: serde_json::Value) -> ApiResponse<serde_json::Value> {
        match method {
            "user/get" => self.handle_get_user(params).await,
            "user/create" => self.handle_create_user(params).await,
            "gateway/routes" => self.handle_get_routes(params).await,
            "gateway/route/create" => self.handle_create_route(params).await,
            "menu/list" => self.handle_get_menus(params).await,
            "menu/create" => self.handle_create_menu(params).await,
            _ => ApiResponse::error(format!("未知方法: {}", method), 404),
        }
    }

    async fn handle_get_user(&self, params: serde_json::Value) -> ApiResponse<serde_json::Value> {
        if let Some(user_id) = params.get("user_id").and_then(|v| v.as_i64()) {
            // 这里应该调用实际的用户管理模块
            ApiResponse::success(serde_json::json!({
                "id": user_id,
                "username": "test_user",
                "email": "test@example.com",
                "status": "active"
            }))
        } else {
            ApiResponse::error("缺少user_id参数".to_string(), 400)
        }
    }

    async fn handle_create_user(&self, params: serde_json::Value) -> ApiResponse<serde_json::Value> {
        // 这里应该调用实际的用户管理模块
        ApiResponse::success(serde_json::json!({
            "id": 1,
            "username": params.get("username").and_then(|v| v.as_str()).unwrap_or(""),
            "email": params.get("email").and_then(|v| v.as_str()).unwrap_or(""),
            "status": "active"
        }))
    }

    async fn handle_get_routes(&self, _params: serde_json::Value) -> ApiResponse<serde_json::Value> {
        // 这里应该调用实际的网关服务模块
        ApiResponse::success(serde_json::json!([]))
    }

    async fn handle_create_route(&self, params: serde_json::Value) -> ApiResponse<serde_json::Value> {
        // 这里应该调用实际的网关服务模块
        ApiResponse::success(serde_json::json!({
            "id": 1,
            "rule_name": params.get("rule_name").and_then(|v| v.as_str()).unwrap_or(""),
            "target_url": params.get("target_url").and_then(|v| v.as_str()).unwrap_or(""),
            "is_active": true
        }))
    }

    async fn handle_get_menus(&self, _params: serde_json::Value) -> ApiResponse<serde_json::Value> {
        // 这里应该调用实际的菜单管理模块
        ApiResponse::success(serde_json::json!([]))
    }

    async fn handle_create_menu(&self, params: serde_json::Value) -> ApiResponse<serde_json::Value> {
        // 这里应该调用实际的菜单管理模块
        ApiResponse::success(serde_json::json!({
            "id": 1,
            "menu_name": params.get("menu_name").and_then(|v| v.as_str()).unwrap_or(""),
            "menu_path": params.get("menu_path").and_then(|v| v.as_str()).unwrap_or(""),
            "is_visible": true
        }))
    }
}

// 事件监听器
pub struct EventListener {
    message_bus: Arc<MessageBus>,
}

impl EventListener {
    pub fn new(message_bus: Arc<MessageBus>) -> Self {
        Self { message_bus }
    }

    // 监听特定类型的事件
    pub fn listen_for_events(&self, event_type: &str) -> impl StreamExt<Item = ModuleMessage> {
        let receiver = self.message_bus.channel(event_type);
        let stream = BroadcastStream::new(receiver);
        stream.filter_map(|msg| std::future::ready(msg.ok()))
    }

    // 处理用户相关事件
    pub async fn handle_user_events(&self) {
        let mut stream = Box::pin(self.listen_for_events("users"));
        
        while let Some(message) = stream.next().await {
            match message {
                ModuleMessage::UserCreated(user) => {
                    println!("用户创建事件: {:?}", user);
                    // 这里可以触发其他模块的响应
                }
                ModuleMessage::UserUpdated(user) => {
                    println!("用户更新事件: {:?}", user);
                }
                ModuleMessage::UserDeleted(user_id) => {
                    println!("用户删除事件: {}", user_id);
                }
                _ => {}
            }
        }
    }

    // 处理网关相关事件
    pub async fn handle_gateway_events(&self) {
        let mut stream = Box::pin(self.listen_for_events("gateway"));
        
        while let Some(message) = stream.next().await {
            match message {
                ModuleMessage::RouteRuleCreated(rule) => {
                    println!("路由规则创建事件: {:?}", rule);
                }
                ModuleMessage::RouteRuleUpdated(rule) => {
                    println!("路由规则更新事件: {:?}", rule);
                }
                ModuleMessage::RouteRuleDeleted(rule_id) => {
                    println!("路由规则删除事件: {}", rule_id);
                }
                ModuleMessage::ConfigUpdated(config) => {
                    println!("配置更新事件: {:?}", config);
                }
                _ => {}
            }
        }
    }
}

// 服务注册中心
#[derive(Debug, Clone)]
pub struct ServiceRegistry {
    services: Arc<RwLock<HashMap<String, ServiceInfo>>>,
}

impl ServiceRegistry {
    pub fn new() -> Self {
        Self {
            services: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    // 注册服务
    pub fn register_service(&self, service_info: ServiceInfo) {
        let mut services = self.services.write().unwrap();
        services.insert(service_info.service_name.clone(), service_info);
    }

    // 注销服务
    pub fn unregister_service(&self, service_name: &str) {
        let mut services = self.services.write().unwrap();
        services.remove(service_name);
    }

    // 获取服务信息
    pub fn get_service(&self, service_name: &str) -> Option<ServiceInfo> {
        let services = self.services.read().unwrap();
        services.get(service_name).cloned()
    }

    // 获取所有服务
    pub fn get_all_services(&self) -> Vec<ServiceInfo> {
        let services = self.services.read().unwrap();
        services.values().cloned().collect()
    }

    // 健康检查
    pub async fn health_check(&self) -> Vec<ServiceHealthResult> {
        let services: Vec<ServiceInfo> = {
            let services_guard = self.services.read().unwrap();
            services_guard.values().cloned().collect()
        };
        
        let mut results = Vec::new();
        
        for service in services {
            let client = reqwest::Client::new();
            let result = client
                .get(&service.health_check_url)
                .timeout(std::time::Duration::from_secs(5))
                .send()
                .await;
                
            let status = match result {
                Ok(response) if response.status().is_success() => "healthy".to_string(),
                Ok(response) => format!("unhealthy: {}", response.status()),
                Err(e) => format!("error: {}", e),
            };
            
            results.push(ServiceHealthResult {
                service_name: service.service_name.clone(),
                status,
                timestamp: OffsetDateTime::now_utc().unix_timestamp(),
            });
        }
        
        results
    }
}

// 通信管理器
pub struct CommunicationManager {
    message_bus: Arc<MessageBus>,
    service_registry: Arc<ServiceRegistry>,
    rpc_server: Arc<RpcServer>,
    event_listener: Arc<EventListener>,
}

impl CommunicationManager {
    pub fn new() -> Self {
        let message_bus = Arc::new(MessageBus::new());
        let service_registry = Arc::new(ServiceRegistry::new());
        let rpc_server = Arc::new(RpcServer::new(message_bus.clone()));
        let event_listener = Arc::new(EventListener::new(message_bus.clone()));

        Self {
            message_bus,
            service_registry,
            rpc_server,
            event_listener,
        }
    }

    // 启动通信系统
    pub async fn start(&self) {
        // 启动事件监听器
        let event_listener = self.event_listener.clone();
        tokio::spawn(async move {
            event_listener.handle_user_events().await;
        });

        let event_listener = self.event_listener.clone();
        tokio::spawn(async move {
            event_listener.handle_gateway_events().await;
        });

        println!("通信系统已启动");
    }

    // 获取消息总线
    pub fn get_message_bus(&self) -> Arc<MessageBus> {
        self.message_bus.clone()
    }

    // 获取服务注册中心
    pub fn get_service_registry(&self) -> Arc<ServiceRegistry> {
        self.service_registry.clone()
    }

    // 获取RPC客户端
    pub fn create_rpc_client(&self, base_url: String) -> RpcClient {
        RpcClient::new(base_url)
    }
}